

// 异步工作器的实现
// 双缓冲区的实现思路
#ifndef __M_LOPPER_H__
#define __M_LOPPER_H__

#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <atomic>

namespace xglog
{
    // 通过回调函数设置对数据的处理方式
    using Func = std::function<void(Buffer &)>;

    // 异步日志器的模式
    // 安全模式：固定大小的缓冲区，写满了就会阻塞
    // 非安全模式：支持一直写入,缓冲区可扩增
    enum class AsyncType
    {
        ASYNC_SAFE,
        ASYNC_UNSAFE,
    };

    // 异步工作器
    class AsyncWorker
    {
    public:
        using ptr = std::shared_ptr<AsyncWorker>;
        AsyncWorker(const Func& callback, AsyncType type = AsyncType::ASYNC_UNSAFE)
            : _stop(false),
              _callback(callback),
              _thread(std::thread(&xglog::AsyncWorker::threadEntry, this)),
              _type(type)
        {
        }

        ~AsyncWorker()
        {
            stop();
        }
        void stop()
        {
            _stop = true;
            // 唤醒消费线程
            _cond_con.notify_all();
            // 等待线程退出
            _thread.join();
        }
        // 向生产缓冲区中添加数据
        void push(const char *data, size_t len)
        {
            // 如果生产缓冲区剩余空间不足，则阻塞住
            std::unique_lock<std::mutex> lock(_mtx);
            // 如果是安全模式才可能会发生阻塞,非安全模式直接进行写入缓冲区操作
            if (_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return _buf_pro.writeAbleSize() >= len; });
            // 走到这里说明生产缓冲区中可以添加数据了
            _buf_pro.push(data, len);
            // 接下来可以唤醒消费者线程
            _cond_con.notify_one();
        }

    private:
        // 线程入口函数
        // 对消费缓冲区中的数据进行处理,处理完后,重置消费缓冲区,交换缓冲区
        void threadEntry()
        {
            while (1)
            {
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    if (_stop == true && _buf_pro.empty())
                        break;
                    // 如果生产缓冲区中有数据,就交换缓冲区,并唤醒消费线程;如果没有,则一直等待
                    // 或者如果该日志器要停止时,唤醒对应的线程
                    _cond_con.wait(lock, [&]()
                                   { return _stop == true || !_buf_pro.empty(); });
                    // 交换缓冲区
                    _buf_con.swap(_buf_pro);
                    // 唤醒生产者
                    // 如果是安全模式才可能会发生阻塞,非安全模式直接进行写入缓冲区操作
                    if (_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 执行回调函数,完成对消费缓冲区数据的处理
                _callback(_buf_con);
                // 重置消费缓冲区
                _buf_con.reset();
            }
        }

    private:
        // 该异步日志器的类型(安全or不安全)
        AsyncType _type;
        std::atomic<bool> _stop; // 当前日志器是否停止运行
        Buffer _buf_pro;         // 生产缓冲区
        Buffer _buf_con;         // 消费缓冲区
        std::mutex _mtx;
        std::condition_variable _cond_pro; // 生产者等待队列
        std::condition_variable _cond_con; // 消费者等待队列
        std::thread _thread;
        Func _callback; // 回调函数,处理数据的处理(消费)
    };
}

#endif